// -*-mode:c++; coding:utf-8-*-

#include <boost/foreach.hpp>

#include <column_name_parser.hpp>
#include <performance_measurement.hpp>

#include "global.hpp"
#include "table_schema.hpp"
#include "table_schema_manager.hpp"
#include "kv_scanner_manager.hpp"
#include "naming_strategy.hpp"
#include "root_meta_tablet_client.hpp"
#include "table_manager.hpp"
#include "tablet.hpp"
#include "kv.hpp"

#include "service_provider.hpp"

namespace xbase
{
	int service_provider::update_table_schema(const TableInfo &p_info)
	{
		// 通过table_info创建相应的table_schema，并更新到管理器中
		table_schema::pointer _schema = this->create_schema(p_info);
		global::the_schema_manager->update_or_add(_schema);

		LOG(INFO, "schema added or updated:" << _schema->to_string());

		return 0;
	}
	
	int service_provider::load_tablet(const LoadTabletReq &p_req,
					  LoadTabletResp &p_resp,
					  TabletSplitReport &p_split_report)
	{
		// 设置部分确定的响应值
		p_resp.tablet_name = p_req.tablet.tablet_name();
		p_resp.server_address.host_name = global::the_address_info->m_hostname;
		p_resp.server_address.port = global::the_address_info->m_port;
		p_resp.server_address.start_code = global::the_address_info->m_startcode;
		p_resp.error_code = PERR_SUCCESS;

		if(global::stop_request)
		{
			LOG(WARN, "stop requested, do not load tablet " << p_req.tablet.to_string());
			p_resp.error_code = error::service_stopped;
			return 0;
		}

		// 找到table对象
		table::pointer _table = global::the_table_manager->get_table(p_req.table.table_name);
		if (! _table)	// 若没有则创建
		{
			_table.reset(new table(p_req.table.table_name));
			global::the_table_manager->add_table(_table);
		}

		// 检查是否已经加载
		bool _is_loaded = _table->is_loaded(p_req.tablet.tablet_name());
		if (_is_loaded)	// 已经加载
		{
			// 是很严重的错误，子表重复加载了
			LOG(FATAL, "load failed, tablet "
			    << p_req.tablet.tablet_name()
			    << " already exists[TODO: check tablet status in root meta tablet]");
			
			// 已经加载，则返回加载成功，避免再次加载
			//
			// TODO: 此时应该检查根子表的内容，并更新子表状态
			//
			// TODO: 返回更确切的错误号
			p_resp.error_code = PERR_SUCCESS;
		}
		else		// 没有加载，则加载，并添加到表中
		{
			tablet::pointer _tablet = tablet::load(global::the_conf->xbase_root_path, p_req.tablet);
			if (_tablet)
			{
				// 如果是根子表，则更新全局的根子表位置信息
				if (_tablet->is_root_meta_tablet()) {
					global::the_root_meta_manager->update(*global::the_address_info);
					_table->add_tablet(_tablet);

					LOG(INFO, "root meta tablet loaded.");
				} else {
					// 如果不是根子表，则连接根子表服务器，将信息写入根元子表
					if (_tablet->is_online())
					{
						// 根子表位置无效，则更新根子表
						const int retry_times = global::the_conf->retry_times;
						int retry_count = 0;
						while((! global::the_root_meta_manager->valid())
						      && (retry_count++ < retry_times))
						{
							LOG(ERROR, "service_provider::load_tablet root meta address is invalid,"
							    << " querying.");
							this->query_root_meta_address();
							
							// 等待几秒钟，期望根子表可以更新
							boost::this_thread::sleep(boost::posix_time::seconds(global::the_conf->wait_seconds_for_master_reply));
						}
						if (! global::the_root_meta_manager->valid()) {
							LOG(ERROR, "service_provider::load_tablet root meta address is still invalid after "
							    << retry_times << " retry(ies)");
							p_resp.error_code = PERR_UNKNOWN; // TODO: add error code
							return 0;
						}
						protocol::server_address _addr = global::the_root_meta_manager->get();
						try {
							root_meta_tablet_client _client(_addr.m_hostname, _addr.m_port);
							if (! _client.is_open()) {
								LOG(ERROR, "service_provider::load_tablet connect to root meta server failed.");
								global::the_root_meta_manager->set_invalid();
								this->query_root_meta_address(); // may be out of date
								p_resp.error_code = PERR_UNKNOWN;
								return 0;
							}

							std::string log_message = "tablet_server: loaded by tablet server ";
							log_message += global::the_address_info->to_string();
							bool _ok = _client.update_tablet(_tablet->tablet_name(),
											 _tablet->get_tablet_info(),
											 *global::the_address_info,
											 global::the_address_info->m_startcode,
											 log_message);
							if (! _ok) {
								LOG(ERROR, "service_provider::load_tablet updating tablet info failed.");
								global::the_root_meta_manager->set_invalid();
								this->query_root_meta_address(); // may be out of date
								p_resp.error_code = PERR_UNKNOWN;
								return 0;
							}
						} catch (const std::exception &e) {
							LOG(ERROR, "service_provider::load_tablet update tablet info failed: "
							    << e.what());
							global::the_root_meta_manager->set_invalid();
							this->query_root_meta_address(); // may be out of date
							p_resp.error_code = PERR_UNKNOWN;
							return 0;
						}

						// 写入成功，则保存好子表
						p_resp.error_code = PERR_SUCCESS;
						_table->add_tablet(_tablet);
						LOG(INFO, "load_tablet tablet " << _tablet->tablet_name() << " is loaded.");
					}
					else if (_tablet->is_split())
					{
						LOG(INFO, "tablet " << _tablet->tablet_name() << " is split.");
						const int ret = update_split_tablet_info(_tablet, p_split_report);
						if(ret < 0) // update failed
						{
							p_resp.error_code = PERR_UNKNOWN; // TODO: add error code
							_tablet->clean_split_path();
							LOG(ERROR, "service_provider::load_tablet update split tablet info failed");
						}
						else
						{
							p_resp.error_code = PERR_SUCCESS;
							LOG(INFO, "load_table: tablet "
							    << _tablet->tablet_name()
							    << " is split successfully.");
						}
					}
					else
					{
						p_resp.error_code = PERR_UNKNOWN;
						LOG(ERROR, "load_table: unexpected tablet status "
						    << _tablet->to_string() << ".");
					}
				}
			}
			else
			{
				LOG(ERROR, "loading tablet " << p_req.tablet.tablet_name() << " failed.");
				// TODO: 设置更精确的错误号
				p_resp.error_code = PERR_UNKNOWN;
			}
		}
		return 0;
	}

	int service_provider::unload_tablet(const RemoveTabletReq &req,
					    RemoveTabletResp &resp)
	{
		resp.tablet_name = req.tablet_name;

		const std::string _table_name = naming_strategy::get_table_name_from_tablet_name(req.tablet_name);
		table::pointer _table = global::the_table_manager->get_table(_table_name);
		if (! _table)
		{
			LOG(ERROR, "service_provider::unload_tablet table "
			    << _table_name
			    << " does not exists");
			resp.error_code = PERR_TABLE_NOT_EXISTS;
			return 0;
		}
		tablet::pointer _tablet = _table->get_tablet(req.tablet_name);
		if (! _tablet)
		{
			LOG(ERROR, "service_provider::unload_tablet tablet "
			    << req.tablet_name
			    << " does not exists");
			resp.error_code = PERR_TABLET_NOT_EXISTS;
			return 0;
		}

		//
		// 按bigtable中的优化，在下线子表时，首先做一次次要紧
		// 缩，并保持服务状态（可读写），可能需要较长的时间；完
		// 成后再将子表设为关闭；然后再做一次次要紧缩操作，可以
		// 确定这一次紧缩很快；然后再写根子表，并且置服务器地址
		// 为空，使得主服务器在加载前不需要分类日志，再真正的删除
		// 该子表。
		//
		// 这样的好处是1、执行次要紧缩可以加快加载子表操作；2、
		// 先以服务状态做较长时间的紧缩，再以关闭状态做较短时间
		// 的紧缩。这两点都能使子表处于不可用状态的时间减少，从
		// 而使子表的移动对服务没有任何影响！
		// 

		// 1. force minor compact
		_tablet->do_minor_compact_forced();

		// 2. 将该子表锁定，不准它改变状态.同时也阻止了对该表的读写、紧缩等操作
		{
			// 不长期持有写锁
			//Guard<ReadWriteLock::WriteLock> guard(_tablet->lock().write_lock());
			boost::unique_lock<boost::shared_mutex> guard(_tablet->lock());
			_tablet->set_online(false); // 设置为下线状态
		}

		// 3. do another fast minor compact
		_tablet->do_minor_compact_forced();

		// 4. be sure no users
		const int retry_times = global::the_conf->retry_times;
		int retry_count = 0;
		while((! _tablet->not_used()) // still in use
		      && (retry_count++ < retry_times))
		{
			LOG(INFO, "service_provider::unload_tablet tablet "
			    << _tablet->tablet_name()
			    << " still in use, wait for a while...");
			boost::this_thread::sleep(boost::posix_time::seconds(global::the_conf->wait_seconds_for_tablet_idle));
		}
		if(! _tablet->not_used()) // still in use
		{
			LOG(ERROR , "service_provider::unload_tablet tablet "
			    << _tablet->tablet_name()
			    << " still in use, unload failed");
			resp.error_code = PERR_UNKNOWN;	// PERR_TABLET_IN_USE;
			return 0;
		}

		// 5. do acturally unload
		std::string log_message = "tablet_server: unloaded by tablet server ";
		log_message += global::the_address_info->to_string();
		log_message += " because of remove-tablet-request from master";
		const int ret = this->unload_tablet(_tablet, log_message);
		if(ret < 0)
		{
			LOG(ERROR, "service_provider::unload_tablet unload failed");
			resp.error_code = PERR_UNKNOWN;
			return 0;
		}

		resp.error_code = PERR_SUCCESS;
		return 0;
	}

	// 将指定子表删除，并更新根子表中，将子表状态设置为子表的当前状
	// 态，子表的服务器地址清空(避免主服务器再来分类该服务器的日
	// 志)，起始码清空
	//
	// 操作前必须要将子表关闭，并确保所有操作都已经被应用
	int service_provider::unload_tablet(const tablet::pointer &tablet,
					    const std::string &log_message)
	{
		//
		// 删除子表
		//
		table::pointer table = global::the_table_manager->get_table(tablet->table_name());
		if(! table)
		{
			LOG(ERROR, "service_provider::unload_tablet get table "
			    << tablet->table_name() << "  failed");
			return -1;
		}

		const size_t num = table->remove_tablet(tablet);
		LOG_IF_ELSE((num != 1),
			    WARN,
			    "service_provider::unload_tablet remove tablet "
			    << tablet->tablet_name() << " failed(" << num << ")",
			    INFO,
			    "service_provider::unload_tablet remove tablet "
			    << tablet->tablet_name() << " ok");
		// TODO: 暂时忽略这个返回值，如果发现真有num != 1的情况，再作处理！

		if(tablet->is_root_meta_tablet()) // 如果是根子表，则不需要更新写根子表中的信息，并且需要将根子表位置设为无效
		{
			global::the_root_meta_manager->set_invalid();
			LOG(WARN, "service_provider::unload_tablet root meta tablet unloaded: " << log_message);
			return 0;
		}

		//
		// 更新根子表
		// 

		// 根子表位置无效，则更新根子表
		const int retry_times = global::the_conf->retry_times;
		int retry_count = 0;
		while((! global::the_root_meta_manager->valid())
		      && (retry_count++ < retry_times))
		{
			LOG(ERROR, "service_provider::unload_tablet root meta address is invalid,"
			    << " querying...");
			this->query_root_meta_address();
			
			// 等待几秒钟，期望根子表可以更新
			boost::this_thread::sleep(boost::posix_time::seconds(global::the_conf->wait_seconds_for_master_reply));
		}
		if (! global::the_root_meta_manager->valid())
		{
			LOG(ERROR, "service_provider::unload_tablet root meta address is still invalid after "
			    << retry_times << " retry(ies)");
			return -1;
		}
		
		protocol::server_address _addr = global::the_root_meta_manager->get();
		try {
			root_meta_tablet_client client(_addr.m_hostname, _addr.m_port);
			if (! client.is_open())
			{
				LOG(ERROR, "service_provider::unload_tablet connect to root meta server failed");
				global::the_root_meta_manager->set_invalid();
				this->query_root_meta_address(); // may be out of date
				return -1;
			}
			const bool ok = client.update_tablet(tablet->tablet_name(),
							     tablet->get_tablet_info(),
							     protocol::server_address(), // must clear these fields
							     0,
							     log_message);
			if(! ok)
			{
				LOG(ERROR, "service_provider::unload_tablet update tablet info failed");
				return -1;
			}
			LOG(DEBUG, "service_provider::unload_tablet update tablet info ok");
		}
		catch (const std::exception &e)
		{
			LOG(ERROR, "service_provider::unload_tablet update tablet info failed:"
			    << e.what());
			global::the_root_meta_manager->set_invalid();
			this->query_root_meta_address(); // may be out of date
			return -1;
		}
		return 0;
	}

	int service_provider::scan_open(ScanOpenReq &req, ScanOpenResp &resp)
	{
		MEASUREMENT_OF_TIME("service_provider::scan_open", 0);

		//ScanOpenReq req = p_req; // make mutable copy
		
		resp.tablet_name = req.tablet_name;
		resp.scanner_id = INVALID_SCAN_ID;

		if(req.column_names.empty())
		{
			LOG(ERROR, "service_provider::scan_open empty column list");
			resp.error_code = PERR_TABLET_NOT_EXISTS;
			return 0;
		}

		// 最新时间戳一定大于或等于最老时间戳，否则直接返回EOS
		if(req.latest_timestamp < req.oldest_timestamp)
		{
			LOG(DEBUG, "service_provider::scan_open null timestamp range, send eos back");
			resp.error_code = PERR_END_OF_SCAN_RECORD;
			return 0;
		}

		const std::string _table_name = this->get_table_name(req.tablet_name);
		table::pointer _table = global::the_table_manager->get_table(_table_name);
		if (! _table)
		{
			LOG(ERROR, "scan_open failed: table " << _table_name << " doesn't exists.");
			resp.error_code = PERR_TABLE_NOT_EXISTS;
			return 0;
		}
		
		//
		// validate scan context against table schema
		//

		table_schema::pointer schema = global::the_schema_manager->get_schema(_table_name);
		if(! schema)
		{
			LOG(ERROR, "service_provider::do_scan_open table schema for "
			    << _table_name << " does not exist");
			resp.error_code = PERR_TABLE_NOT_EXISTS; // TODO: add error code
			return 0;
		}

		tablet::pointer _tablet = _table->get_tablet(req.tablet_name);
		if (! _tablet)
		{
			LOG(ERROR, "scan_open failed: tablet " << req.tablet_name << " doesn't exists.");
			resp.error_code = PERR_TABLET_NOT_EXISTS;
			return 0;
		}
		if(! _tablet->is_online())
		{
			LOG(ERROR, "service_provider::scan_open tablet "
			    << req.tablet_name << " is not onlining");
			resp.error_code = PERR_TABLET_CLOSED;
			return 0;
		}

		// 检查扫描行范围，若不在子表范围，则直接返回；若起始行
		// 为空或小于不为空的当前子表起始行，则替换为当前子表起
		// 始行；若结束行为空或大于当前的结束行，则替换为当前不
		// 为空的结束行
		//
		// 这样提高效率，同时可避免扫描超出指定的范围
		//
		// 注意：在其它地方构造扫描参数的时候，也要按照该方式进行！
		// 
		// 即取交集: {scan range} ^ {tablet range}
		if(req.start_row_name.empty()
		   || ((_tablet->start_row().empty() == false)
		       && (req.start_row_name < _tablet->start_row())))
		{
			LOG(DEBUG, "service_provider::scan_open replace scan start row \""
			    << req.start_row_name << "\" to tablet start row \""
			    << _tablet->start_row() << "\"");
			req.start_row_name = _tablet->start_row();
		}
		if(req.end_row_name.empty()
		   || ((_tablet->end_row().empty() == false)
		       && (req.end_row_name >= _tablet->end_row())))
		{
			LOG(DEBUG, "service_provider::scan_open replace scan end row \""
			    << req.end_row_name << "\" to tablet end row \""
			    << _tablet->end_row() << "\"");
			req.end_row_name = _tablet->end_row();
		}

		//
		// create scanner
		// 

		kv_scan_context::pointer _context = this->create_context(req);
		if(_context->has_bad_family())
		{
			resp.scanner_id = INVALID_SCAN_ID;
			resp.error_code = PERR_FAMILY_NOT_EXISTS;
			LOG(ERROR, "service_provider::scan_open context has bad families: "
			    << _context->to_string());
			return 0;
		}
		
		tablet_scanner::pointer _scanner = _tablet->get_scanner(_context);
		if (! _scanner)
		{
			resp.scanner_id = INVALID_SCAN_ID;
			resp.error_code = PERR_UNKNOWN;
			LOG(ERROR, "service_provider::scan_open get scanner failed.");
		}
		else if(_scanner->eos())
		{
			resp.scanner_id = INVALID_SCAN_ID;
			resp.error_code = PERR_END_OF_SCAN_RECORD;
			LOG(DEBUG, "service_provider::scan_open got empty scanner, return eos");
		}
		else
		{
			global::the_scanner_manager->add_scanner(_scanner);
			
			resp.scanner_id = _scanner->id();
			resp.error_code = PERR_SUCCESS;
			LOG(TRACE, "scan_open success on tablet " << req.tablet_name << ".");
		}
		return 0;
	}

	int service_provider::scan_next(const ScanNextReq &req, ScanNextResp &resp)
	{
		MEASUREMENT_OF_TIME("service_provider::scan_next", req.scanner_id);

		resp.scanner_id = req.scanner_id;
		resp.tablet_name = req.tablet_name;
		
		// 若子表分裂后，会将相应的子表扫描器全部关闭，那么这里将找不到扫描器，因此会返回“扫描器不存在”错误
		tablet_scanner::pointer _scanner = global::the_scanner_manager->get_scanner(req.scanner_id);
		if(! _scanner)	// scanner does not exists
		{
			LOG(ERROR, "scan_next failed: scanner doesn't exist.");
			resp.error_code = PERR_SCANNER_NOT_EXISTS;
			return 0;
		}

		// check tablet name
		if (_scanner->tablet_name() != req.tablet_name)
		{
			LOG(ERROR, "scan_next failed: scanner's tablet_name doesn't match.");
			resp.error_code = PERR_SCANNER_NOT_EXISTS;
			return 0;
		}

		// update timestamp
		_scanner->update_timestamp();

		if(! _scanner->eos())
		{
			_scanner->fill_records_by_row(resp.records);
		}
		if(resp.records.empty())
		{
			resp.error_code = PERR_END_OF_SCAN_RECORD;
			LOG(TRACE, "service_provider::scan_next succeeded: touched eos.");
		}
		else
		{
			resp.error_code = PERR_SUCCESS;
			LOG(TRACE, "service_provider::scan_next succeeded: got "
			    << resp.records.size() << " records.");
		}

		return 0;
	}

	int service_provider::scan_close(const ScanCloseReq &req, ScanCloseResp &resp)
	{
		MEASUREMENT_OF_TIME("service_provider::scan_close", req.scanner_id);
		
		resp.tablet_name = req.tablet_name;
		resp.scanner_id = req.scanner_id;

		{
			tablet_scanner::pointer _scanner = global::the_scanner_manager->get_scanner(req.scanner_id);
			if(! _scanner)	// scanner does not exists
			{
				LOG(ERROR, "scan_close failed: scanner doesn't exist.");
				resp.error_code = PERR_SCANNER_NOT_EXISTS;
				return 0;
			}
			
			// check tablet name
			if (_scanner->tablet_name() != req.tablet_name)
			{
				LOG(ERROR, "scan_close failed: scanner's tablet_name doesn't match.");
				resp.error_code = PERR_SCANNER_NOT_EXISTS;
				return 0;
			}
		}

		// 确认了一切后，才能将其关闭
		global::the_scanner_manager->close_scanner(req.scanner_id);
		resp.error_code = PERR_SUCCESS;

		return 0;
	}
	
	int service_provider::write(request::pointer request_ptr,
				    protocol::write_response_v2 &resp)
	{
		const boost::asio::const_buffer body_buffer = request_ptr->get_body_buffer();

		protocol::write_request_v2_for_tabletserver req;
		protocol::iarchive in(body_buffer);
		if(! req.load(in))
		{
			resp.set_error_code(error::packet_decode_failed);
			return 0;
		}

		const std::size_t skip_bytes = (sizeof(uint32_t) + // error_code
						sizeof(uint32_t) + req.m_tablet_name.size()); // tablet_name
		const boost::asio::const_buffer mutation_data = body_buffer + skip_bytes;
		
		//resp.error_code = PERR_SUCCESS;
		const std::string &tablet_name = req.m_tablet_name;

		// 
		// 1. validation
		//

		// 1.1 found table & tablet and check if it is online
		const std::string _table_name = this->get_table_name(tablet_name);
		table::pointer _table = global::the_table_manager->get_table(_table_name);
		if (! _table)
		{
			LOG(ERROR, "service_provider::write table "
			    << _table_name << " doesn't exist.");
			resp.set_error_code(error::table_not_exists);
			return 0;
		}
		tablet::pointer _tablet = _table->get_tablet(tablet_name);
		if (! _tablet)
		{
			LOG(ERROR, "service_provider::write tablet "
			    << tablet_name << " doesn't exist.");
			resp.set_error_code(error::tablet_not_exists);
			return 0;
		}
		if(! _tablet->is_online())
		{
			LOG(DEBUG, "service_provider::write tablet "
			    << tablet_name << " is not onlining");
			resp.set_error_code(error::tablet_closed);
			return 0;
		}

		//
		// 3. write into tablet
		// 

		// 先写入子表中，因为写子表失败即可返回失败并保证没有数
		// 据被写入；写成功后，若写日志失败，则认为文件系统不可
		// 用，此时子表服务器退出退可（客户端自然会检测到写入失败）。
		//
		// 但日志文件尾部可能会形成不完整的记录，需要在分类日志
		// 的时候忽略掉。
		error::error_code ec = error::ok;
		int ret = _tablet->add(req.m_mutation_data,
				       mutation_data,
				       ec); // 原子的，要么全部写成功，要么全部写失败，不能保证就挂掉程序
		if (ret < 0)	// writing failed
		{
			LOG(DEBUG, "service_provider::write write to tablet "
			    << tablet_name << " error("
			    << ec << ")");
			resp.set_error_code(ec);
			return 0;
		}
		// success

		// 写日志放在了子表的写操作中

		//
		// 5. set response fields
		// 
		
		resp.set_error_code(error::ok); // 现在，只要返回成功，即表示所有数据都写入成功
		return 0;
	}
	
	int service_provider::batch_write(request::pointer request_ptr,
					  protocol::batch_write_response_v2 &resp)
	{
		const boost::asio::const_buffer body_buffer = request_ptr->get_body_buffer();

		resp.m_success_row = 0;

		protocol::batch_write_request_v2_for_tabletserver req;
		protocol::iarchive in(body_buffer);
		if(! req.load(in))
		{
			resp.set_error_code(error::packet_decode_failed);
			return 0;
		}

		const std::string &tablet_name = req.m_tablet_name;

		// 
		// 1. validation
		//

		// 1.1 found table & tablet and check if it is online
		const std::string _table_name = this->get_table_name(tablet_name);
		table::pointer _table = global::the_table_manager->get_table(_table_name);
		if (! _table)
		{
			LOG(ERROR, "service_provider::batch_write table "
			    << _table_name << " doesn't exist.");
			resp.set_error_code(error::table_not_exists);
			return 0;
		}
		tablet::pointer _tablet = _table->get_tablet(tablet_name);
		if (! _tablet)
		{
			LOG(ERROR, "service_provider::batch_write tablet "
			    << tablet_name << " doesn't exist.");
			resp.set_error_code(error::tablet_not_exists);
			return 0;
		}
		if(! _tablet->is_online())
		{
			LOG(DEBUG, "service_provider::batch_write tablet "
			    << tablet_name << " is not onlining");
			resp.set_error_code(error::tablet_closed);
			return 0;
		}

		// 这里暂时就不对表结构进行检测了，因为写入子表后，在子
		// 表级按组分类时，也会对列族进行检查，确保了存在的列族
		// 和不包含正则表达式的数据才能写成功。

		std::size_t i = 0;
		BOOST_FOREACH(const protocol::row_data_v2_for_tabletserver &row, req.m_rows)
		{
			error::error_code ec = error::ok;
			const int ret = _tablet->add(row.m_mutation_data,
						     row.m_raw_data,
						     ec);
			if(ret < 0) // failed
			{
				LOG(DEBUG, "service_provider::batch_write write "
				    << i << "th row to tablet "
				    << tablet_name << " error("
				    << ec << ")");
				resp.set_error_code(ec);
				resp.m_success_row = i;
				return 0;
			}
			else
			{
				++i;
				LOG(TRACE, "service_provider::batch_write written "
				    << i << "th row");
			}
		}
		
		LOG(TRACE, "service_provider::batch_write writen "
		    << i << " row(s).");
		resp.set_error_code(error::ok);
		resp.m_success_row = i;

		return 0;
	}

	int service_provider::read(const ReadReq &req, ReadResp &resp)
	{
		resp.tablet_name = req.tablet_name;
		resp.records.clear();

		// 
		// 1. validation
		const std::string &_tablet_name = req.tablet_name;
		const std::string _table_name = naming_strategy::get_table_name_from_tablet_name(_tablet_name);
		table::pointer _table = global::the_table_manager->get_table(_table_name);
		if (! _table)
		{
			LOG(ERROR, "read failed: table " << _table_name << " does not exists.");
			resp.error_code = PERR_TABLE_NOT_EXISTS;
			return 0;
		}

		table_schema::pointer _schema = global::the_schema_manager->get_schema(_table_name);
		if (! _schema)
		{
			LOG(ERROR, "read failed: table schema " << _table_name << " does not exists.");
			resp.error_code = PERR_TABLE_NOT_EXISTS;
			return 0;
		}
		// check if family exists
		column_name_parser parser(req.key.column_name);
		if ((! parser.is_any_column()) // 不是任意列，才检查列族
		    && (! _schema->has_family(parser.family_name())))
		{
			LOG(ERROR, "read failed: family "
			    << parser.family_name()
			    << " does not exists in table "
			    << _table_name);
			resp.error_code = PERR_FAMILY_NOT_EXISTS;
			return 0;
		}
		tablet::pointer _tablet = _table->get_tablet(_tablet_name);
		if (! _tablet)
		{
			LOG(ERROR, "read failed: tablet " << _tablet_name << " does not exists.");
			resp.error_code = PERR_TABLET_NOT_EXISTS;
			return 0;
		}
		if(! _tablet->is_online())
		{
			LOG(ERROR, "service_provider::read tablet "
			    << req.tablet_name << " is not onlining");
			resp.error_code = PERR_TABLET_CLOSED;
			return 0;
		}
		// check if row name is in tablet range
		if (((! _tablet->start_row().empty()) && (req.key.row_name < _tablet->start_row())) ||
		    ((! _tablet->end_row().empty()) && (req.key.row_name >= _tablet->end_row())))
		{
			LOG(ERROR, "read failed: query row " << req.key.row_name << " does not in the range ("
			    << _tablet->start_row() << ","
			    << _tablet->end_row() << ") of tablet"
			    << _tablet_name << ".");
			resp.error_code = PERR_TABLET_NOT_EXISTS;
			return 0;
		}

		//
		// 2. convert read op to a scan op
		ScanOpenReq _scan_req;
		_scan_req.tablet_name = _tablet_name;
		_scan_req.start_row_name = req.key.row_name;
		const std::string end_row = req.key.row_name + " ";
		// 进一步缩小可能的扫描范围，但要保证不能超过子表的结束
		// 行，否则对分裂后的子表，可能会读取到不属于它的记录
		if(! _tablet->end_row().empty())
		{
			_scan_req.end_row_name = std::min(_tablet->end_row(), end_row);
		}
		else
		{
			_scan_req.end_row_name = end_row;
		}
		_scan_req.column_names.push_back(req.key.column_name);
		_scan_req.latest_timestamp = req.key.time_stamp;
		_scan_req.oldest_timestamp = NO_TIMESTAMP;
		_scan_req.ttl = 0;
		_scan_req.version_number = 1;
		_scan_req.options = 0; // filter deleted
		_scan_req.num_row = 1;

		//
		// 3. scanning for reading
		kv_scan_context::pointer _context = this->create_context(_scan_req);
		if(_context->has_bad_family())
		{
			resp.error_code = PERR_FAMILY_NOT_EXISTS;
			LOG(ERROR, "service_provider::read context has bad families: "
			    << _context->to_string());
			return 0;
		}

		tablet_scanner::pointer _scanner = _tablet->get_scanner(_context);
		if(! _scanner)
		{
			LOG(ERROR, "read failed: open scanner failed.");
			resp.error_code = PERR_UNKNOWN;
			return 0;
		}
		if(_scanner->eos())
		{
			LOG(DEBUG, "service_provider::read no record to read");
			resp.error_code = PERR_SUCCESS;
			return 0;
		}

		//
		// 4. 检查读取回来的是不是指定行，若不是，则没有这一行
		// 数据，则不读取
		if((*_scanner->get().m_key->row()) != req.key.row_name)
		{
			LOG(DEBUG, "service_provider::read no record belong to row "
			    << dump_string_char(req.key.row_name));
			resp.error_code = PERR_SUCCESS;
			return 0;
		}
		_scanner->fill_records_by_row(resp.records);
		LOG(DEBUG, "service_provider::read "
		    << resp.records.size()
		    << " records readed");
		resp.error_code = PERR_SUCCESS;

		return 0;
	}

	int service_provider::query_root_meta_address()
	{
		response::pointer query(response::factory_type::construct());
		SET_HEADER(query->get_header(),
			   ROOT_META_REQ,
			   0,
			   0);

		RootMetaReq _req;
		query->get_header().m_data_length = query->add_body(_req);
		query->save_header();

		LOG_ENV(const int ret = ) global::the_master_session->write(query);
		LOG_IF((ret < 0), ERROR,
		       "send query root meta request failed");

		return ret;
	}

	int service_provider::update_split_tablet_info(const tablet::pointer &tablet,
						       TabletSplitReport &p_split_report)
	{
		const protocol::tablet_info _parent = tablet->get_tablet_info();
		const protocol::tablet_info _split_a = tablet->get_split_a_info();
		const protocol::tablet_info _split_b = tablet->get_split_b_info();

		// 根子表位置无效，则更新根子表
		const int retry_times = global::the_conf->retry_times;
		int retry_count = 0;
		while((! global::the_root_meta_manager->valid())
		      && (retry_count++ < retry_times))
		{
			LOG(ERROR, "service_provider::update_split_tablet_info root meta address is invalid,"
			    << " querying.");
			this->query_root_meta_address();
			
			// 等待几秒钟，期望根子表可以更新
			boost::this_thread::sleep(boost::posix_time::seconds(global::the_conf->wait_seconds_for_master_reply));
		}
		if (! global::the_root_meta_manager->valid())
		{
			LOG(ERROR, "service_provider::update_split_tablet_info root meta address is still invalid after "
			    << retry_times << " retry(ies)");
			return -1;
		}
		
		protocol::server_address _addr = global::the_root_meta_manager->get();
		try {
			root_meta_tablet_client _client(_addr.m_hostname, _addr.m_port);
			if (! _client.is_open())
			{
				LOG(ERROR, "service_provider::update_split_tablet_info connect to root meta server failed.");
				global::the_root_meta_manager->set_invalid();
				this->query_root_meta_address(); // may be out of date
				return -1;
			}

			std::string log_message = "tablet_server: generated by spliting on tablet server ";
			log_message += global::the_address_info->to_string();

			// 这里先写子子表信息，都成功后再更改父子表信
			// 息；避免父子表信息更新了，而子子表信息更新不
			// 成功而导致的子表丢失问题。

			// write split_a info
			bool _ok = _client.update_tablet(_split_a.tablet_name(),
							 _split_a,
							 log_message);
			if (! _ok)
			{
				LOG(ERROR, "service_provider::update_split_tablet_info"
				    << " writing split_a tablet info failed.");
				global::the_root_meta_manager->set_invalid();
				this->query_root_meta_address(); // may be out of date
				return -1;
			}
			LOG(TRACE, "service_provider::update_split_tablet_info split_a tablet info updated.");

			// write split_b info
			_ok = _client.update_tablet(_split_b.tablet_name(),
						    _split_b,
						    log_message);
			if (! _ok)
			{
				LOG(ERROR, "service_provider::update_split_tablet_info"
				    << " writing split_b tablet info failed.");
				global::the_root_meta_manager->set_invalid();
				this->query_root_meta_address(); // may be out of date
				return -1;
			}
			LOG(TRACE, "service_provider::update_split_tablet_info split_b tablet info updated.");

			
			log_message = "tablet_server: split by mid-row \"";
			log_message += _split_a.m_end_row;
			log_message += "\" on tablet server ";
			log_message += global::the_address_info->to_string();
			
			// update parent tablet info
			_ok = _client.update_tablet(tablet->tablet_name(),
						    _parent,
						    // 表明是这个子表服务器分裂的
						    *global::the_address_info, 
						    global::the_address_info->m_startcode,
						    _split_a,
						    _split_b,
						    log_message);
			if (! _ok)
			{
				LOG(ERROR, "service_provider::update_split_tablet_info updating tablet info failed.");
				global::the_root_meta_manager->set_invalid();
				this->query_root_meta_address(); // may be out of date
				return -1;
			}
			LOG(TRACE, "service_provider::update_split_tablet_info parent tablet info updated.");
		}
		catch (const std::exception &e)
		{
			LOG(ERROR, "service_provider::update_split_tablet_info update tablet info failed:"
			    << e.what());
			global::the_root_meta_manager->set_invalid();
			this->query_root_meta_address(); // may be out of date
			return -1;
		}
		
		// split success， 不加入在线子表列表中
		p_split_report.original_tablet = _parent;
		p_split_report.top_tablet = _split_a;
		p_split_report.bottom_tablet = _split_b;
		
		return 0;
	}

	void service_provider::report_tablet_server_quit()
	{
		response::pointer report(response::factory_type::construct());
		SET_HEADER(report->get_header(),
			   TABLET_SERVER_QUIT,
			   0,
			   0);

		ServerAddress addr;
		addr = *global::the_address_info;
		TabletServerQuit quit_report(addr);
	
		report->get_header().m_data_length = report->add_body(quit_report);
		report->save_header();
		LOG_ENV(const int ret = ) global::the_master_session->write(report);
		LOG_IF((ret < 0), ERROR,
		       "send quit report failed");
		
		// 等待几秒，确保消息发送完毕
		//boost::this_thread::sleep(boost::posix_time::seconds(global::the_conf->wait_seconds_for_master_reply));
	}

	int service_provider::list_tablet(std::vector<TabletInfo> &tablet_infos)
	{
		std::vector<table::pointer> table_list;
		std::vector<tablet::pointer> tablet_list;
		global::the_table_manager->list_table(table_list);
		BOOST_FOREACH(const table::pointer &table, table_list)
		{
			table->list_tablet(tablet_list);
			BOOST_FOREACH(const tablet::pointer &ptr, tablet_list)
			{
				TabletInfo info;
				info = ptr->get_tablet_info();
				tablet_infos.push_back(info);
			}
			tablet_list.clear();
		}
		return 0;
	}

	int service_provider::unload_all_tablet(const std::string &log_message)
	{
		std::vector<table::pointer> table_list;
		std::vector<tablet::pointer> tablet_list;
		global::the_table_manager->list_table(table_list);
		BOOST_FOREACH(const table::pointer &table, table_list)
		{
			table->list_tablet(tablet_list);
			BOOST_FOREACH(const tablet::pointer &ptr, tablet_list)
			{
				// 大致按照unload_tablet的步骤来
				ptr->do_minor_compact_forced();
				{
					boost::unique_lock<boost::shared_mutex> guard(ptr->lock());
					ptr->set_online(false); // 设置为下线状态
				}
				ptr->do_minor_compact_forced();
				LOG_ENV(int ret = ) unload_tablet(ptr, log_message);
				LOG_IF_ELSE((ret < 0),
					    ERROR,
					    "unload tablet " << ptr->tablet_name() << " failed, ignore it",
					    INFO,
					    "tablet " << dump_string_char(ptr->tablet_name()) << " unloaded");
			}
			tablet_list.clear();
		}
		return 0;
	}

	// === new interface ===

	void service_provider::do_load_tablet(request::pointer request_ptr,
					      response::pointer response_ptr)
	{
		LOG(TRACE, "service_provider::do_load_tablet");

		// set response header
		response_ptr->get_header().m_magic = XBASE_PROTOCOL_MAGIC;
		response_ptr->get_header().m_version = TABLET_SERVER_VERSION;
		response_ptr->get_header().m_type = LOAD_TABLET_RESP;
		response_ptr->get_header().m_sequence_number = request_ptr->get_header().m_sequence_number;
		response_ptr->get_header().m_data_length = 0; // 初始设置为0

		const boost::asio::const_buffer buffer = request_ptr->get_body_buffer();
		ByteBuffer bb(readable_bb(buffer));
		BinIArchive in(bb);
		LoadTabletReq req;
		in & req;

		// TMP: 为找加载空子表名的错误而加！
		if(req.tablet.table_name.empty())
		{
			LOG(ERROR, "empty table name!");
			LOG(ERROR, "service_provider::do_load_tablet load_tabletreq: "
			    << req);
		}
		else
		{
			LOG(DEBUG, "service_provider::do_load_tablet load_tabletreq: "
			    << req);
		}
	
		// 用最新的表结构信息更新当前的
		update_table_schema(req.table);
	
		LoadTabletResp _resp;
		TabletSplitReport _split_report;
		_split_report.top_tablet.tablet_id = INVALID_TABLET_ID; // set flag
		load_tablet(req, _resp, _split_report);

		// 添加协议包内容，同时设置包长度
		response_ptr->get_header().m_data_length = response_ptr->add_body(_resp);
		response_ptr->save_header();
		
		// if has split, send split report
		if ( (_resp.error_code == PERR_SUCCESS) &&
		     (_split_report.top_tablet.tablet_id != INVALID_TABLET_ID)) {
			response addition_response;
			addition_response.get_header().m_magic = XBASE_PROTOCOL_MAGIC;
			addition_response.get_header().m_version = TABLET_SERVER_VERSION;
			addition_response.get_header().m_type = TABLET_SPLIT_REPORT;
			addition_response.get_header().m_sequence_number = request_ptr->get_header().m_sequence_number;
			addition_response.get_header().m_data_length = addition_response.add_body(_split_report);
			addition_response.save_header();

			response_ptr->append_response(addition_response);
		}
	}

	void service_provider::do_remove_tablet(request::pointer request_ptr,
						response::pointer response_ptr)
	{
		LOG(TRACE, "service_provider::do_remove_tablet");

		// set response header
		response_ptr->get_header().m_magic = XBASE_PROTOCOL_MAGIC;
		response_ptr->get_header().m_version = TABLET_SERVER_VERSION;
		response_ptr->get_header().m_type = REMOVE_TABLET_RESP;
		response_ptr->get_header().m_sequence_number = request_ptr->get_header().m_sequence_number;
		response_ptr->get_header().m_data_length = 0; // 初始设置为0

		RemoveTabletReq _req;
		ByteBuffer in_bb(readable_bb(request_ptr->get_body_buffer()));
		BinIArchive in(in_bb); in & _req;
	
		RemoveTabletResp _resp;
		unload_tablet(_req, _resp);

		response_ptr->get_header().m_data_length = response_ptr->add_body(_resp);
		response_ptr->save_header();
	}

	void service_provider::do_root_meta(request::pointer request_ptr)
	{
		LOG(TRACE, "service_provider::do_root_meta");
		
		RootMetaResp _resp;
		ByteBuffer in_bb(readable_bb(request_ptr->get_body_buffer()));
		BinIArchive in(in_bb); in & _resp;

		if (_resp.error_code == PERR_SUCCESS)
		{
			// 看地址是否有效
			if (_resp.server_address.host_name.size() == 0 ||
			    _resp.server_address.port == 0 ||
			    _resp.server_address.start_code == 0)
			{
				LOG(ERROR, "invalid address for root meta, skip it.");
				return;
			}

			protocol::server_address addr;
			addr.m_hostname = _resp.server_address.host_name;
			addr.m_port = _resp.server_address.port;
			addr.m_startcode = _resp.server_address.start_code;
			global::the_root_meta_manager->update(addr);
			
			LOG(INFO, "root meta info updated:" << _resp << ".");
		}
		else
		{
			LOG(ERROR, "root meta _response error(" << _resp.error_code << "):"
			    << protocolErrorMessage [_resp.error_code] << ".");
		}
	}

	void service_provider::do_table_info(request::pointer request_ptr)
	{
		LOG(TRACE, "service_provider::do_table_info");

		TableInfoResp _resp;
		ByteBuffer in_bb(readable_bb(request_ptr->get_body_buffer()));
		BinIArchive in(in_bb); in & _resp;

		if (_resp.error_code == PERR_SUCCESS)
		{
			update_table_schema(_resp.table);
		}
		else
		{
			LOG(ERROR, "response packet about table "<< _resp.table.table_name
			    <<" info error(" << _resp.error_code << "):"
			    << protocolErrorMessage [_resp.error_code] << ".");
		}
	}

	void service_provider::do_restart(request::pointer request_ptr)
	{
		LOG(WARN, "service_provider::do_restart");

		// TODO:
		// 尽快退出，避免再写元数据表；暂时采用暴力手段；以后可
		// 在写元数据表的地方加入标志。
		assert(false && "service_provider::do_restart");
	}

	void service_provider::do_unexpected(request::pointer request_ptr)
	{
		LOG(WARN, "MasterMessageHandler::do_unexpected unexpected message type 0x"
		    << std::hex << std::uppercase
		    << request_ptr->get_header().m_type
		    << ".");
		
		// do nothing
	}

	// --- client request ---
	
	void service_provider::do_read(request::pointer request_ptr,
				       response::pointer response_ptr)
	{
		LOG(TRACE, "service_provider::do_read");

		// set response header
		SET_HEADER(response_ptr->get_header(),
			   READ_RESP,
			   request_ptr->get_header().m_sequence_number,
			   0);

		ReadReq _req;
		ByteBuffer in_bb(readable_bb(request_ptr->get_body_buffer()));
		BinIArchive in(in_bb); in & _req;

		ReadResp _resp;
		read(_req, _resp);

		response_ptr->get_header().m_data_length = response_ptr->add_body(_resp);
		response_ptr->save_header();
	}

	void service_provider::do_write(request::pointer request_ptr,
					response::pointer response_ptr)
	{
		LOG(TRACE, "service_provider::do_write");

		SET_HEADER(response_ptr->get_header(),
			   WRITE_RESP,
			   request_ptr->get_header().m_sequence_number,
			   0);
		
		//WriteResp _resp;
		protocol::write_response_v2 _resp;
		write(request_ptr, _resp);

		const std::size_t buf_size = 256;
		char buf[buf_size];
		protocol::oarchive out(buf, buf_size);
		LOG_ENV(bool ok = )
			_resp.save(out);
		assert(ok);

		boost::asio::mutable_buffer b = response_ptr->add_empty_buffer(out.used_buffer_size());
		::memcpy(boost::asio::buffer_cast<char*>(b),
			 buf,
			 boost::asio::buffer_size(b));

		response_ptr->get_header().m_data_length = boost::asio::buffer_size(b);	// response_ptr->add_body(_resp);
		response_ptr->save_header();
	}

	void service_provider::do_batch_write(request::pointer request_ptr,
					      response::pointer response_ptr)
	{
		LOG(TRACE, "service_provider::do_batch_write");

		SET_HEADER(response_ptr->get_header(),
			   BATCH_WRITE_RESP,
			   request_ptr->get_header().m_sequence_number,
			   0);
		
		//BatchWriteResp _resp;
		protocol::batch_write_response_v2 _resp;
		batch_write(request_ptr, _resp);
		
		const std::size_t buf_size = 256;
		char buf[buf_size];
		protocol::oarchive out(buf, buf_size);
		LOG_ENV(bool ok = )
			_resp.save(out);
		assert(ok);

		boost::asio::mutable_buffer b = response_ptr->add_empty_buffer(out.used_buffer_size());
		::memcpy(boost::asio::buffer_cast<char*>(b),
			 buf,
			 boost::asio::buffer_size(b));

		response_ptr->get_header().m_data_length = boost::asio::buffer_size(b);	// response_ptr->add_body(_resp);
		response_ptr->save_header();
	}

	void service_provider::do_scan_open(request::pointer request_ptr,
					    response::pointer response_ptr)
	{
		LOG(TRACE, "service_provider::do_scan_open");

		SET_HEADER(response_ptr->get_header(),
			   SCAN_OPEN_RESP,
			   request_ptr->get_header().m_sequence_number,
			   0);

		ScanOpenReq _req;
		ByteBuffer in_bb(readable_bb(request_ptr->get_body_buffer()));
		BinIArchive in(in_bb); in & _req;

		LOG(TRACE, "service_provider::do_scan_open req: " << _req.to_string());
		
		ScanOpenResp _resp;
		scan_open(_req, _resp);
		
		response_ptr->get_header().m_data_length = response_ptr->add_body(_resp);
		response_ptr->save_header();
	}

	void service_provider::do_scan_next(request::pointer request_ptr,
					    response::pointer response_ptr)
	{
		LOG(TRACE, "service_provider::do_scan_next");

		SET_HEADER(response_ptr->get_header(),
			   SCAN_NEXT_RESP,
			   request_ptr->get_header().m_sequence_number,
			   0);

		ScanNextReq _req;
		ByteBuffer in_bb(readable_bb(request_ptr->get_body_buffer()));
		BinIArchive in(in_bb); in & _req;

		ScanNextResp _resp;
		scan_next(_req, _resp);

		response_ptr->get_header().m_data_length = response_ptr->add_body(_resp);
		response_ptr->save_header();
	}
	
	void service_provider::do_scan_close(request::pointer request_ptr,
					     response::pointer response_ptr)
	{
		TRACE("service_provider::do_scan_close");

		SET_HEADER(response_ptr->get_header(),
			   SCAN_CLOSE_RESP,
			   request_ptr->get_header().m_sequence_number,
			   0);
		
		ScanCloseReq _req;
		ByteBuffer in_bb(readable_bb(request_ptr->get_body_buffer()));
		BinIArchive in(in_bb); in & _req;
		
		ScanCloseResp _resp;
		scan_close(_req, _resp);

		response_ptr->get_header().m_data_length = response_ptr->add_body(_resp);
		response_ptr->save_header();
	}

	void service_provider::do_list_tablet(request::pointer request_ptr,
					      response::pointer response_ptr)
	{
		LOG(TRACE, "service_provider::do_list_tablet");

		SET_HEADER(response_ptr->get_header(),
			   LIST_TABLET_RESP,
			   request_ptr->get_header().m_sequence_number,
			   0);

		list_tablet_resp _resp;
		list_tablet(_resp.tablet_infos);
		
		response_ptr->get_header().m_data_length = response_ptr->add_body(_resp);
		response_ptr->save_header();
	}

	void service_provider::construct_heartbeat(response::pointer message)
	{
		SET_HEADER(message->get_header(),
			   HEARTBEAT_REQ,
			   0,
			   0);

		// update load info
		global::the_table_manager->get_stat(global::the_load_info->m_tablet_number,
						    global::the_load_info->m_memory_table_size);
		global::the_load_info->m_cached_block_size = sstable_block_cache_manager::instance().cached_block_size();
		global::the_load_info->m_cached_block_number = sstable_block_cache_manager::instance().cached_block_number();
		global::the_load_info->update_meminfo();
		global::the_load_info->update_cpuinfo();
		global::the_load_info->update_loadinfo();
		
		LOG(INFO, "tablet_server load info:" << *global::the_load_info);

		ServerAddress addr;
		addr = *global::the_address_info;
		HeartBeatReq _heartbeat(ServerInfo(addr,
						   *global::the_load_info));
		// reset stat
		global::the_load_info->reset();

		message->get_header().m_data_length = message->add_body(_heartbeat);
		message->save_header();
	}

	void service_provider::construct_report(response::pointer report)
	{
		SET_HEADER(report->get_header(),
			   TABLET_SERVER_REPORT,
			   0,
			   0);

		ServerAddress addr;
		addr = *global::the_address_info;
		TabletServerReport _report(addr);
		report->get_header().m_data_length = report->add_body(_report);
		report->save_header();
	}

} // namespace
